package net.ufrog.common.redis;

import io.codis.jodis.JedisResourcePool;
import io.codis.jodis.RoundRobinJedisPool;
import net.ufrog.common.Logger;
import net.ufrog.common.cache.Cache;
import net.ufrog.common.cache.CacheKey;
import net.ufrog.common.exception.ServiceException;
import net.ufrog.common.redis.serializers.Serializer;
import redis.clients.jedis.Jedis;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Stream;

/**
 * @author ultrafrog, ufrog.net@gmail.com
 * @version 0.1, 2017-01-16
 * @since 0.1
 */
public class CodisImpl implements Cache {

    private static final byte[] EX = "EX".getBytes();
    private static final byte[] NX = "NX".getBytes();
    private static final byte[] XX = "XX".getBytes();

    private Serializer serializer;
    private JedisResourcePool jedisResourcePool;

    /**
     * 构造函数
     *
     * @param cacheKey 代码
     */
    public CodisImpl(CacheKey cacheKey) {
        String server = cacheKey.getConfig("server", null);
        String proxy = cacheKey.getConfig("proxy.dir");
        String serializer = cacheKey.getConfig("serializer");
        Integer timeout = Integer.valueOf(cacheKey.getConfig("timeout", "30000"));
        jedisResourcePool = RoundRobinJedisPool.create().curatorClient(server, timeout).zkProxyDir(proxy).build();
        try {
            this.serializer = (Serializer) Class.forName(serializer).newInstance();
            Logger.info("initialize '%s' codis cache success!", server);
        } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
            throw new ServiceException(e.getMessage(), e);
        }
    }

    @Override
    public void add(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            if (jedis.exists(key.getBytes())) return;
            jedis.set(key.getBytes(), serializer.serialize(value), NX, EX, expiration);
        }
    }

    @Override
    public boolean safeAdd(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            if (jedis.exists(key.getBytes())) return false;
            jedis.set(key.getBytes(), serializer.serialize(value), NX, EX, expiration);
            return true;
        } catch (Throwable e) {
            Logger.error(e.getMessage(), e);
            return false;
        }
    }

    @Override
    public void set(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            byte[] nxxx = jedis.exists(key.getBytes()) ? XX : NX;
            jedis.set(key.getBytes(), serializer.serialize(value), nxxx, EX, expiration);
        }
    }

    @Override
    public boolean safeSet(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            byte[] nxxx = jedis.exists(key.getBytes()) ? XX : NX;
            jedis.set(key.getBytes(), serializer.serialize(value), nxxx, EX, expiration);
            return true;
        } catch (Throwable e) {
            Logger.error(e.getMessage(), e);
            return false;
        }
    }

    @Override
    public void replace(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            if (!jedis.exists(key.getBytes())) return;
            jedis.set(key.getBytes(), serializer.serialize(value), XX, EX, expiration);
        }
    }

    @Override
    public boolean safeReplace(String key, Object value, int expiration) {
        try (Jedis jedis = getJedis()) {
            if (!jedis.exists(key.getBytes())) return false;
            jedis.set(key.getBytes(), serializer.serialize(value), XX, EX, expiration);
            return true;
        } catch (Throwable e) {
            Logger.error(e.getMessage(), e);
            return false;
        }
    }

    @Override
    public Object get(String key) {
        try (Jedis jedis = getJedis()) {
            if (jedis.exists(key.getBytes())) return serializer.deserialize(jedis.get(key.getBytes()));
            return null;
        }
    }

    @Override
    public Map<String, Object> get(String... keys) {
        try (Jedis jedis = getJedis()) {
            Map<String, Object> values = new LinkedHashMap<>(keys.length);
            Stream.of(keys).forEach(key -> {
                Object value = jedis.exists(key.getBytes()) ? serializer.deserialize(jedis.get(key.getBytes())) : null;
                values.put(key, value);
            });
            return values;
        }
    }

    @Override
    public long incr(String key, int by) {
        try (Jedis jedis = getJedis()) {
            if (!jedis.exists(key.getBytes())) return -1;
            return jedis.incrBy(key.getBytes(), by);
        }
    }

    @Override
    public long decr(String key, int by) {
        try (Jedis jedis = getJedis()) {
            if (!jedis.exists(key.getBytes())) return -1;
            return jedis.decrBy(key.getBytes(), by);
        }
    }

    @Override
    public void delete(String key) {
        try (Jedis jedis = getJedis()) {
            jedis.del(key.getBytes());
        }
    }

    @Override
    public boolean safeDelete(String key) {
        try (Jedis jedis = getJedis()) {
            jedis.del(key.getBytes());
            return true;
        } catch (Throwable e) {
            Logger.error(e.getMessage(), e);
            return false;
        }
    }

    @Override
    public void clear() {
        try (Jedis jedis = getJedis()) {
            jedis.flushDB();
        }
    }

    @Override
    public void stop() {
        Logger.warn("cannot stop server.");
    }

    /**
     * Jedis实例<br>
     * 如果外部引用请确保最后关闭<br>
     * 连接池连接有限
     *
     * @return jedis instance
     */
    public Jedis getJedis() {
        return jedisResourcePool.getResource();
    }
}
